package org.dxw.common;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.dxw.common.enums.TableChangeEnum;
import org.dxw.vo.TableChangeMsg;
import org.mockito.internal.util.StringUtil;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @Author dxw
 * @Date 2023/11/19 14:46
 * @Description 监听表变动的事件，路由给相关的服务
 * @warn 不负责线程安全，线程安全由相关服务负责
 */
@Component
@Slf4j
public class TableChangeHandler {

    /**
     * 表变动就需要消费的数据，因为不知道哪些数据更新了，所以全表（或者是大部分）数据更新
     * 资源消耗大，所以做了一个优化，每隔一段时间消费，且把多个消息合并为一个消息消费
     */
    private ConcurrentLinkedQueue<TableChangeMsg> tableAllDataUpdate;

    private ConcurrentLinkedQueue<TableChangeMsg> failConsumeMsgList;

    @Async
    public void addMsg(TableChangeMsg tableChangeMsg){
        //后续可以使用消息中间件，防止程序重启时，丢失消息

        //1 如果知道某些记录更新，直接执行业务逻辑
        if(CollUtil.isNotEmpty(tableChangeMsg.getRowDOList())){
            consumerHasRowListMsg(tableChangeMsg);
        }else{
            //2 否则放入缓存
            tableAllDataUpdate.offer(tableChangeMsg);
        }
    }

    private void consumerHasRowListMsg(TableChangeMsg tableChangeMsg){
        if (CollUtil.isEmpty(tableChangeMsg.getRowDOList())) return;
        Set<TableChangeEnum> tableChangeEnums = TableChangeEnum.getConsumerSet(tableChangeMsg);
        for (TableChangeEnum tableChangeEnum : tableChangeEnums) {
            if (tableChangeEnum.isSupportRowsUpdate()) {
                try {
                    Boolean apply = (Boolean) tableChangeEnum.getConsumer().apply(tableChangeMsg.getRowDOList());
                    if (apply != null && !apply) {
                        saveFailMsg(Arrays.asList(tableChangeMsg), tableChangeEnum, "业务方法执行结果返回false");
                    }
                } catch (Exception e) {
                    log.warn("消费消息:{}失败", tableChangeMsg.getMsgUUID(), e);
                    saveFailMsg(Arrays.asList(tableChangeMsg), tableChangeEnum, e.getMessage());
                    //后续可以把消费失败的消息入库
                }
            }
        }
    }

    /**
     *
     */
    @Scheduled(fixedRate = 60*1000)
    public void consumerDelay(){
        List<TableChangeMsg> tableChangeMsgs = new ArrayList<>();
        try{
            while(!tableAllDataUpdate.isEmpty()){
                tableChangeMsgs.add(tableAllDataUpdate.poll());
            }

            Map<TableChangeEnum, List<TableChangeMsg>> consumerTableChangeMsgMap = multiMsg2ConsumerList(tableChangeMsgs);

            for (Map.Entry<TableChangeEnum, List<TableChangeMsg>> entry : consumerTableChangeMsgMap.entrySet()) {
                TableChangeEnum tableChangeEnum = entry.getKey();
                List<TableChangeMsg> tableChangeMsgs1 = entry.getValue();
                try{
                    Boolean apply = (Boolean)tableChangeEnum.getConsumer().apply(null);
                    if(apply!=null && !apply){
                        saveFailMsg(tableChangeMsgs1, tableChangeEnum, "业务方法执行返回false");
                    }
                }catch (Exception e){
                    saveFailMsg(tableChangeMsgs1, tableChangeEnum, e.getMessage());
                }
            }
        }catch (Exception e){
            log.warn("表更新处理器消费消息发生异常", e);
            //这里先这样粗暴处理
            saveFailMsg(tableChangeMsgs, null, e.getMessage());
        }

    }


    @Scheduled(fixedRate = 60*1000)
    public void consumerfailMsgDelay(){
        List<TableChangeMsg> tableChangeMsgs = new ArrayList<>();
        try{
            while(!failConsumeMsgList.isEmpty()){
                tableChangeMsgs.add(failConsumeMsgList.poll());
            }

            Map<TableChangeEnum, List<TableChangeMsg>> consumerTableChangeMsgMap = multiMsg2ConsumerList(tableChangeMsgs);
            for (Map.Entry<TableChangeEnum, List<TableChangeMsg>> entry : consumerTableChangeMsgMap.entrySet()) {
                TableChangeEnum tableChangeEnum = entry.getKey();
                List<TableChangeMsg> tableChangeMsgs1 = entry.getValue();
                try{
                    Boolean apply = (Boolean)tableChangeEnum.getConsumer().apply(null);
                    if(apply!=null && !apply){
                        saveFailMsg(tableChangeMsgs1, tableChangeEnum, "业务方法执行返回false");
                    }
                }catch (Exception e){
                    saveFailMsg(tableChangeMsgs1, tableChangeEnum, e.getMessage());
                }
            }
        }catch (Exception e){
            log.warn("表更新处理器消费消息发生异常", e);
            //这里先这样粗暴处理
            saveFailMsg(tableChangeMsgs, null, e.getMessage());
        }
    }


    private void saveFailMsg(List<TableChangeMsg> tableChangeMsgs, TableChangeEnum tableChangeEnum, String failMsg){
        log.warn("消费者{},消费失败, 失败到消息为:{}", tableChangeEnum==null?null:tableChangeEnum.getName(), tableChangeMsgs.stream().map(x->x.getMsgUUID()).collect(Collectors.joining()), failMsg);
        for (TableChangeMsg tableChangeMsg : tableChangeMsgs) {
            tableChangeMsg.consumerFail(tableChangeEnum, failMsg);
        }
        for (TableChangeMsg tableChangeMsg : tableChangeMsgs) {
            failConsumeMsgList.offer(tableChangeMsg);
        }
    }

    private Map<TableChangeEnum, List<TableChangeMsg>> multiMsg2ConsumerList(List<TableChangeMsg> tableChangeMsgs){
        //相同的消费者只需要消费一次，所以使用map合并
        Map<TableChangeEnum, List<TableChangeMsg>> consumerTableChangeMsgMap = new HashMap<>();
        for (TableChangeMsg tableChangeMsg : tableChangeMsgs) {
            Set<TableChangeEnum> consumerSet = consumerSetRemoveFail(tableChangeMsg);
            for (TableChangeEnum tableChangeEnum : consumerSet) {
                List<TableChangeMsg> msgList = consumerTableChangeMsgMap.get(tableChangeEnum);
                if(msgList == null){
                    msgList = new LinkedList<>();
                    consumerTableChangeMsgMap.put(tableChangeEnum, msgList);
                }
                msgList.add(tableChangeMsg);
            }
        }
        return consumerTableChangeMsgMap;
    }

    private Set<TableChangeEnum> consumerSetRemoveFail (TableChangeMsg tableChangeMsg){
        Set<TableChangeEnum> consumerSet = TableChangeEnum.getConsumerSet(tableChangeMsg);
        Iterator<TableChangeEnum> iterator = consumerSet.iterator();
        while(iterator.hasNext()){
            TableChangeEnum tableChangeEnum = iterator.next();
            for (TableChangeMsg.FailConsumer failConsumer : tableChangeMsg.getFailConsumerList()) {
                if(failConsumer.getName().equals(tableChangeEnum.getName()) && failConsumer.getFailCount()>tableChangeEnum.getMaxFailCount()){
                    iterator.remove();
                }
            }
        }
        return consumerSet;
    }


}
